package com.jacquesh;

import com.jacquesh.model.CoffeeShop;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

public class FlinkDemo0009 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");

        final OutputTag<CoffeeShop> outputTag = new OutputTag<CoffeeShop>("side-output") {
        };
        SingleOutputStreamOperator dataStream = stream

                .flatMap(new RichFlatMapFunction<String, CoffeeShop>() {

                    @Override
                    public void flatMap(String s, Collector<CoffeeShop> collector) {
                        try {
                            String sp[] = s.split(",");

                            collector.collect(CoffeeShop.builder()
                                    .id(Long.parseLong(sp[0]))
                                    .date(format.parse(sp[1]))
                                    .sales(Integer.parseInt(sp[2]))
                                    .build());
                        } catch (Exception ex) {

                            System.out.println(ex);

                        }
                    }
                })
                .process(new ProcessFunction<CoffeeShop, CoffeeShop>() {

                    @Override
                    public void processElement(CoffeeShop s, Context context, Collector<CoffeeShop> collector) {
                        try {
                            if (s.getId() % 2 == 0) {
                                collector.collect(s);
                            } else {
                                context.output(outputTag, s);
                            }

                        } catch (Exception ex) {
                            System.out.println(ex);

                        }
                    }
                });

        dataStream.getSideOutput(outputTag).print("outputTag>>>>>>>>>>>>>");
        dataStream.print("stream>>>>>>>>>>>>>");

        environment.execute("process demo");
    }
}
